package come.whvc

object Test10 {
  def main(args: Array[String]): Unit = {
    val logs = List(
      "2024-06-01|ERROR|spark job failed",
      "2024-06-01|INFO|spark job started",
      "2024-06-01|ERROR|executor lost",
      "2024-06-01|ERROR|spark job failed"
    )

    val errorMessages = logs.filter(line => line.split("\\|")(1) == "ERROR")
      .map(line => line.split("\\|")(2))
    //filter过滤：对每条日志line，用split("\\|")按|分割，得到一个数组[日期, 级别, 消息]
    // 判断数组第 2 个元素（索引1）是否为"ERROR"，保留符合条件的日志
    //map提取消息：对过滤后的每条 ERROR 日志，再次按|分割，取数组第 3 个元素（索引2），即日志的具体消息内容（如"spark job failed"）
    //结果errorMessages是一个仅包含 ERROR 消息的列表：List("spark job failed", "executor lost", "spark job failed")

    val wordCounts = errorMessages.flatMap(msg => msg.split(" "))  // 步骤1：拆分单词并展平
      .groupBy(identity) // 步骤3：按单词分组
      .mapValues(words => words.size) // 步骤4：统计每组数量（词频）
    //flatMap拆分单词：对每条消息msg，用split(" ")按空格拆分成单词数组（如"spark job failed"拆分为["spark", "job", "failed"]）
    // flatMap将多个数组 “展平” 为单个数组，结果为：List("spark", "job", "failed", "executor", "lost", "spark", "job", "failed")
    //groupBy(identity)分组：按单词自身（identity表示 “按元素本身”）分组，相同单词被分到一组，
    // 结果为一个Map：Map("spark" -> List("spark", "spark"), "job" -> List("job", "job"), ...)
    //mapValues统计词频：对分组后的每个值（单词列表），用size计算长度（即出现次数）
    // 最终wordCounts是一个 “单词→次数” 的Map：Map("spark" -> 2, "job" -> 2, "failed" -> 2, "executor" -> 1, "lost" -> 1)

    val topWords = wordCounts.toList  // 转成列表便于排序
      .sortBy(-_._2)  // 按词频降序排序
      .take(3)  // 取前3个元素
    //toList转列表：将Map转为元组列表（每个元组是(单词, 次数)），便于排序
    //sortBy(-_._2)排序：_._2表示元组的第 2 个元素（即次数），-表示 “降序”（从大到小）
    // 排序后结果为：List(("spark", 2), ("job", 2), ("failed", 2), ("executor", 1), ("lost", 1))
    //take(3)取前 3 个：最终topWords为前 3 个高频单词：List(("spark", 2), ("job", 2), ("failed", 2))

    topWords.foreach { case(word, count) =>
      println(s"单词：$word,出现次数：$count")
      //foreach遍历：对topWords中的每个元组，用模式匹配case (word, count)提取单词和次数
      //println输出：用字符串插值s"..."格式化输出
    }
  }
}
